package com.mr;

import java.io.IOException;
import java.util.Date;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 处理rating文件
 */
public class ClearRatingMR {

  public static class ClearRatingMapper extends Mapper<Text,Text,Text, Text> {

    @Override
    protected void map(Text key, Text value, Context context)
        throws IOException, InterruptedException {
      String line = key.toString();
      if(StringUtils.isNotEmpty(line)) {
        String[] arr = line.replaceAll("\"","").split(";");
        if(Integer.parseInt(arr[2]) != 0) {
          arr[1] = MD5Encrypt.encode(arr[1]);
          context.write(new Text(String.join(";", arr)), new Text(""));
        }
      }
    }
  }

  public static class ClearRatingReducer extends Reducer<Text,Text,Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context)
        throws IOException, InterruptedException {
      for (Text value : values) {
        context.write(key, value);
      }
    }
  }

  public static void main(String[] args)
      throws InterruptedException, IOException, ClassNotFoundException {
    Configuration conf = new Configuration();
    org.apache.hadoop.mapreduce.Job job = org.apache.hadoop.mapreduce.Job.getInstance(conf);

    job.setJarByClass(ClearRatingMR.class);

    job.setInputFormatClass(KeyValueTextInputFormat.class);

    job.setMapperClass(ClearRatingMapper.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);
    FileInputFormat.setInputPaths(job, new Path(args[0]));

    job.setReducerClass(ClearRatingReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    FileOutputFormat.setOutputPath(job, new Path(args[1]+"/"+new Date().getTime()));



    job.waitForCompletion(true);
  }
}
